home *** CD-ROM | disk | FTP | other *** search
/ Personal Computer World 2008 February / PCWFEB08.iso / Software / Freeware / Miro 1.0 / Miro_Installer.exe / xulrunner / python / test / framework.py < prev    next >
Encoding:
Python Source  |  2007-11-12  |  5.4 KB  |  166 lines
  1. import unittest
  2.  
  3. import database
  4. import eventloop
  5. import frontend
  6. import app
  7. import downloader
  8. import views
  9. import util
  10. import databaseupgrade
  11. import storedatabase
  12. import subscription
  13. import selection
  14. from time import sleep
  15.  
  16. util.setupLogging()
  17.  
  18. # Generally, all test cases should extend DemocracyTestCase or
  19. # EventLoopTest.  DemocracyTestCase cleans up any database changes you
  20. # might have made, and EventLoopTest provides an API for accessing the
  21. # eventloop in addition to managing the thread pool and cleaning up
  22. # any events you may have scheduled.
  23. # 
  24. # Our general strategy here is to "revirginize" the environment after
  25. # each test, rather than trying to reset applicable pieces of the
  26. # environment before each test. This way, when writing new tests you
  27. # don't have to anticipate what another test may have changed, you
  28. # just have to make sure you clean up what you changed. Usually, that
  29. # is handled transparently through one of these test cases
  30.  
  31. class HadToStopEventLoop(Exception):
  32.     pass
  33.  
  34. class DummyMainFrame:
  35.     def __init__(self):
  36.         self.displays = {}
  37.         self.mainDisplay = "mainDisplay"
  38.         self.channelsDisplay = "channelsDisplay"
  39.         self.collectionDisplay = "collectionDisplay"
  40.         self.videoInfoDisplay = "videoInfoDisplay"
  41.  
  42.     def selectDisplay(self, display, area):
  43.         self.displays[area] = display
  44.  
  45.     def getDisplay(self, area):
  46.         return self.displays.get(area)
  47.  
  48.     def onSelectedTabChange(self, tabType, multiple, guideURL, videoFilename):
  49.         pass
  50.  
  51. class DummyVideoDisplay:
  52.     def fileDuration (self, filename, callback):
  53.         pass
  54.  
  55.     def fillMovieData (self, filename, movie_data, callback):
  56.         pass
  57.  
  58. class DummyController:
  59.     def __init__(self):
  60.         self.selection = selection.SelectionHandler()
  61.         self.frame = DummyMainFrame()
  62.         self.videoDisplay = DummyVideoDisplay()
  63.  
  64. class DemocracyTestCase(unittest.TestCase):
  65.     def setUp(self):
  66.         database.set_thread()
  67.         views.initialize()
  68.         # reset the event loop
  69.         util.chatter = False
  70.         self.oldUtilDotFailed = util.failed
  71.         self.failedCalled = False
  72.         self.utilDotFailedOkay = False
  73.         def newUtilDotFailed(*args, **kwargs):
  74.             if self.utilDotFailedOkay:
  75.                 self.failedCalled = True
  76.             else:
  77.                 print "util.failed called!"
  78.                 print "args: %s kwargs: %s"  % (args, kwargs)
  79.                 import traceback
  80.                 if kwargs.get('withExn'):
  81.                     traceback.print_exc()
  82.                 else:
  83.                     traceback.print_stack()
  84.                 raise Exception("util.failed called")
  85.         util.failed = newUtilDotFailed
  86.         app.controller = DummyController()
  87.  
  88.     def tearDown(self):
  89.         util.chatter = True
  90.  
  91.         # Remove any leftover database
  92.         database.resetDefaultDatabase()
  93.  
  94.         # Remove anything that may have been accidentally queued up
  95.         eventloop._eventLoop = eventloop.EventLoop()
  96.  
  97. class EventLoopTest(DemocracyTestCase):
  98.     def setUp(self):
  99.         DemocracyTestCase.setUp(self)
  100.         self.hadToStopEventLoop = False
  101.  
  102.     def stopEventLoop(self, abnormal = True):
  103.         self.hadToStopEventLoop = abnormal
  104.         eventloop.quit()
  105.  
  106.     def runPendingIdles(self):
  107.         idleQueue = eventloop._eventLoop.idleQueue
  108.         urgentQueue = eventloop._eventLoop.urgentQueue
  109.         while idleQueue.hasPendingIdle() or urgentQueue.hasPendingIdle():
  110.             urgentQueue.processIdles()
  111.             idleQueue.processNextIdle()
  112.  
  113.     def runEventLoop(self, timeout=10, timeoutNormal=False):
  114.         eventloop.threadPoolInit()
  115.         try:
  116.             self.hadToStopEventLoop = False
  117.             timeout = eventloop.addTimeout(timeout, self.stopEventLoop, 
  118.                                            "Stop test event loop")
  119.             eventloop._eventLoop.quitFlag = False
  120.             eventloop._eventLoop.loop()
  121.             if self.hadToStopEventLoop and not timeoutNormal:
  122.                 raise HadToStopEventLoop()
  123.             else:
  124.                 timeout.cancel()
  125.         finally:
  126.             eventloop.threadPoolQuit()
  127.  
  128.     def addTimeout(self,delay, function, name, args=None, kwargs=None):
  129.         eventloop.addTimeout(delay, function, name, args, kwargs)
  130.  
  131.     def addWriteCallback(self, socket, callback):
  132.         eventloop.addWriteCallback(socket, callback)
  133.  
  134.     def removeWriteCallback(self, socket):
  135.         eventloop.removeWriteCallback(socket)
  136.  
  137.     def addIdle(self, function, name, args=None, kwargs=None):
  138.         eventloop.addIdle(function, name, args=None, kwargs=None)
  139.  
  140.     def hasIdles(self):
  141.         return not (eventloop._eventLoop.idleQueue.queue.empty() and
  142.                     eventloop._eventLoop.urgentQueue.queue.empty())
  143.  
  144.     def processThreads(self):
  145.         eventloop._eventLoop.threadPool.initThreads()
  146.         while not eventloop._eventLoop.threadPool.queue.empty():
  147.             sleep(0.05)
  148.         eventloop._eventLoop.threadPool.closeThreads()
  149.  
  150.     def processIdles(self):
  151.         eventloop._eventLoop.idleQueue.processIdles()
  152.         eventloop._eventLoop.urgentQueue.processIdles()
  153.  
  154. class DownloaderTestCase(EventLoopTest):
  155.     def setUp(self):
  156.         EventLoopTest.setUp(self)
  157.         # FIXME: This is kind of ugly
  158.         app.delegate = frontend.UIBackendDelegate()
  159.         downloader.startupDownloader()
  160.  
  161.     def tearDown(self):
  162.         downloader.shutdownDownloader(eventloop.quit)
  163.         self.runEventLoop()
  164.         app.delegate = None
  165.         EventLoopTest.tearDown(self)
  166.